package com.suncreate.zhtz.iot.kafka;

import java.util.Date;

import com.alibaba.fastjson.JSONObject;
import com.suncreate.zhtz.iot.model.*;
import com.suncreate.zhtz.iot.service.*;
import com.suncreate.zhtz.iot.tool.CheckAlarmUtil;
import com.suncreate.zhtz.iot.tool.checkDataUnit;
import org.apache.kafka.clients.consumer.*;
import org.assertj.core.util.DateUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author 孟赟强
 * @date 2021/9/7-7:33
 */
@Component
public class Consumer implements ApplicationRunner {

    @Value("${kafka.consumer.servers}")
    private String servers;
    @Value("${kafka.consumer.enable.auto.commit}")
    private boolean enableAutoCommit;
    @Value("${kafka.consumer.session.timeout}")
    private String sessionTimeout;
    @Value("${kafka.consumer.auto.commit.interval}")
    private String autoCommitInterval;
    @Value("${kafka.consumer.group.id}")
    private String groupId;
    @Value("${kafka.consumer.topics}")
    private String topicsDfault;
    @Value("${kafka.consumer.auto.offset.reset}")
    private String autoOffsetReset;

    @Autowired
    private ScWarningStrategyService scWarningStrategyService;
    @Autowired
    private ScOriginAlarmService scOriginAlarmService;
    @Autowired
    private ScOcrTaohuataiService scOcrTaohuataiService;
    @Autowired
    private ScOcrThalesMotorOilService scOcrThalesMotorOilService;
    @Autowired
    private ScOcrThalesMotorOilspillService scOcrThalesMotorOilspillService;
    @Autowired
    private ScOcrThalesStatusService scOcrThalesStatusService;
    @Autowired
    private ScOcrTimeService scOcrTimeService;
    //订阅主题队列
    final ConcurrentLinkedQueue<String> subscribedTopics = new ConcurrentLinkedQueue<>();
    //策略
    private List<ScWarningStrategy> scWarningStrategys = new ArrayList<>();

    @Override
    public void run(ApplicationArguments args) {

        Properties props = new Properties();
        props.put("bootstrap.servers", servers);
        props.put("group.id", groupId);
        props.put("auto.offset.reset", autoOffsetReset);
        props.put("enable.auto.commit", enableAutoCommit);
        props.put("session.timeout.ms", sessionTimeout);
        props.put("auto.commit.interval.ms", autoCommitInterval);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //1.创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        //2.订阅Topic（最开始从数据库查一遍订阅列表）
//        consumer.subscribe(Arrays.asList("topic_vibration"));
        List<String> topicByDB = scWarningStrategyService.queryTopicnameBySavelocation("kafka");
        //如果策略表里为空，则先订阅默认配置的主题
        if (topicByDB.isEmpty()) {
            consumer.subscribe(Arrays.asList(topicsDfault));
        } else {
            consumer.subscribe(topicByDB);
        }

        //3.轮询
        try {
            while (true) {
                //2000 是超时时间（ms），在该时间内 poll 会等待服务器返回数据
                ConsumerRecords<String, String> records = consumer.poll(2000);
                //打印订阅topic
                System.out.println("已监听的订阅主题：" + consumer.subscription());
                for (ConsumerRecord<String, String> record : records) {
                    //打印offset和value
//                    System.out.printf("offset = %d, value = %s", record.offset(), record.value());
//                    System.out.println();
                    //从kafka里取出数据，判断是哪个设备
                    String value = record.value();
                    String topic = record.topic();
                    System.out.println(topic + "---" + value);
                    //value字符串转JSON对象
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    switch (topic) {
                        //温度 电压电流
                        case "topic_c2000":
                            //倾斜角度
                        case "topic_rtu": {
                            //采集时间
                            Long collectTime = (Long) jsonObject.get("collectTime");
                            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                            Date time = sdf.parse(sdf.format(new Date(collectTime)));

                            for (Object ndmList : jsonObject.getJSONArray("ndmList")) {
                                JSONObject ndm = (JSONObject) ndmList;
                                //通道号
                                String channel = ndm.get("channel").toString();
                                //测量值
                                String num = ndm.get("value").toString();

                                //执行告警判断方法将产生的告警信息（topic、传感器编号、采集时间、采集值）写入数据库
                                this.consume(topic, time, channel, num);
                            }
                            break;
                        }
                    }

                }
                //如果subscribedTopics不为空则说明订阅主题发生改变
                if (!subscribedTopics.isEmpty()) {
                    Iterator<String> iter = subscribedTopics.iterator();
                    List<String> topics = new ArrayList<>();
                    while (iter.hasNext()) {
                        topics.add(iter.next());
                    }
                    //清空订阅队列
                    subscribedTopics.clear();
                    // 重新订阅topic
                    consumer.subscribe(topics);
                }
            }
        } catch (ParseException e) {
            e.printStackTrace();
        } finally {
            //退出应用程序前使用close方法关闭消费者，网络连接和socket也会随之关闭
            consumer.close();
        }
    }

    /**
     * 定时从数据库取测量值并消费
     */
    @Scheduled(fixedRate = 1000 * 10)
    public void consumeByDB() {
        //1.查询所有从数据库中读取的策略
        List<String> topicNames = scWarningStrategyService.queryTopicnameBySavelocation("mysql");
        //2.遍历所有的策略
        for (String topicName : topicNames) {
            String[] tName = topicName.split(",");
            //3.查询上一次更新时间，没有则插入为当前时间，并跳过下面异常逻辑判断
            ScOcrTime scOcrTime = new ScOcrTime();
            scOcrTime.setTablename(tName[0]);
            scOcrTime.setColumnname(tName[1]);
            List<ScOcrTime> scOcrTimes = scOcrTimeService.queryAll(scOcrTime);
            if (scOcrTimes.isEmpty()) {
                scOcrTime.setUpdateTime(new Date());
                scOcrTimeService.insert(scOcrTime);
                break;
            }
            Date startTime = scOcrTimes.get(0).getUpdateTime();
            Date endTime = new Date();

            switch (tName[0]) {
                case "sc_ocr_taohuatai": {//桃花台DVOR状态监测采集器 桃花台DVOR数据监测采集器 桃花台DME数据监测采集器
                    //4.从相关的表中获取这一时间段内的值的集合
                    List<Map<String, Object>> scOcrTaohuatais = scOcrTaohuataiService.queryByDtime(tName[1], startTime,
                            endTime);
                    //5.将异常数据插入告警推送表
                    for (Map<String, Object> scOcrTaohuatai : scOcrTaohuatais) {
                        //测量值
                        String str = (String) scOcrTaohuatai.get(tName[1]);
                        //采集时间
                        Date dTime = (Date) scOcrTaohuatai.get("DTime");

                        //执行告警判断方法将产生的告警信息（topic、传感器编号、采集时间、采集值）写入数据库
                        this.consume(topicName, dTime, tName[1], str);
                    }
                }
                case "sc_ocr_thales_motor_oil": {//泰雷兹雷达状态监测采集器

                }
                case "sc_ocr_thales_motor_oilspill": {//泰雷兹雷达转台电机油位采集器

                }
                case "sc_ocr_thales_status": {//泰雷兹雷达转台电机漏油采集器

                }
                default: {
                    System.out.println("告警策略配置topicName数据库名称有误!");
                }
            }
        }
    }

    /**
     * @param topic   主题
     * @param channel 通道号或者传感器编码
     * @param time    采集时间
     * @param value   采集值
     */
    public void consume(String topic, Date time, String channel, String value) {

        //传感器测试的值，用于判断是否异常
        String str = null;
        BigDecimal num = null;
        if (isNumeric(value)) {
            num = new BigDecimal(value);
        } else {
            str = value;
        }
        //存入scWarningStrategy对象用于查询
        ScWarningStrategy scWarningStrategy = new ScWarningStrategy();
        scWarningStrategy.setTopicname(topic);
        scWarningStrategy.setDataitem(channel);

        //获取所有策略scWarningStrategys
        if (scWarningStrategys.isEmpty()) {
            scWarningStrategys = this.getWarningStrategys();
        }

        for (ScWarningStrategy warningStrategy : scWarningStrategys) {
            if (channel.equals(warningStrategy.getDataitem()) && topic.equals(warningStrategy.getTopicname())) {
                //取出策略判断是否为区间，有区间和非区间两种判断方法
                Map<String, Object> resultMap = new HashMap<>();
                List<ScWarningStrategy> warningStrategys = new ArrayList<>();
                warningStrategys.add(warningStrategy);
                String weakalarm = warningStrategy.getWeakalarm();
                //如果第一个字母为'['或'('，则认为是区间
                if (weakalarm != null && (weakalarm.charAt(0) == '[' || weakalarm.charAt(0) == '(')) {
                    //返回判断结果
                    resultMap = checkDataUnit.checkData(num, warningStrategys);
                } else {
                    //返回判断结果
                    resultMap = CheckAlarmUtil.checkData(str, warningStrategys);
                }

                //如果不为0说明出现异常，则将相关信息写入告警表
                if (!(0 == (Integer) resultMap.get("checkDataStart"))) {
                    ScOriginAlarm scOriginAlarm = new ScOriginAlarm();
                    scOriginAlarm.setAlarmTime(time);
                    scOriginAlarm.setAlarmContent((String) resultMap.get("alarmMessage"));
                    scOriginAlarm.setGeneratedTime(new Date());
                    scOriginAlarm.setAlarmLevel((String) resultMap.get("alarmMessage"));
                    scOriginAlarm.setSensorId(warningStrategy.getDataitem());
                    scOriginAlarm.setEquipId(warningStrategy.getCollector());
                    try {
                        scOriginAlarmService.insert(scOriginAlarm);
                    } catch (Exception e) {
                        System.out.println("数据插入失败！");
                        e.printStackTrace();
                    }
                }
                //打印出消耗的时间
//                Long startTime = Long.valueOf(map.get("startTime"));
//                System.out.println("消耗的时间为：" + (new Date().getTime() - startTime) + "ms");
                break;
            }
        }
    }

    /**
     * 检查topic订阅，10秒查询一次，不同则更新策略
     */
    @Scheduled(fixedRate = 1000 * 10)
    public void checkTopics() {
        // 变更为订阅topic
        List<String> topics = scWarningStrategyService.queryTopicnameBySavelocation("kafka");
        List<String> subTopics = new ArrayList<>();
        for (String subscribedTopic : subTopics) {
            subTopics.add(subscribedTopic);
        }
        //如果两个topic集合不相等
        if (!checkListDiffrent(topics, subTopics)) {
            //更新订阅
            subscribedTopics.addAll(topics);
        }
    }

    /**
     * 定时获取策略表信息并更新
     */
    @Scheduled(fixedRate = 1000 * 10)
    public List<ScWarningStrategy> getWarningStrategys() {
        scWarningStrategys = scWarningStrategyService.queryAll(null);
        return scWarningStrategys;
    }

    /**
     * 判断是否为数字
     *
     * @param str
     * @return
     */
    public boolean isNumeric(String str) {
        if (null == str || "".equals(str)) {
            return false;
        }
        str = str.trim();
        //Integer
        Pattern patternInt = Pattern.compile("^[-\\+]?[\\d]*$");
        //Float或者Double
        Pattern patternFloat = Pattern.compile("^[-\\+]?\\d*[.]\\d+$");
        Matcher isNum = patternInt.matcher(str);
        Matcher isFloat = patternFloat.matcher(str);
        if (isNum.matches() || isFloat.matches()) {
            return true;
        }
        return false;
    }

    /**
     * 使用list自带的sort方法先进行排序，然后转成toString去判断两个集合是否相等
     */
    private static boolean checkListDiffrent(List<String> list, List<String> list1) {
        list.sort(Comparator.comparing(String::hashCode));
        list1.sort(Comparator.comparing(String::hashCode));
        return list.toString().equals(list1.toString());
    }
}
